package com.mojo.web.kafka;

import com.mojo.web.constant.MyConstants;
import com.mojo.web.pojo.User;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * 自定义生产者
 *
 * @author <a href="mailto:sjj@jianzhimao.com">mojo</a>
 * copyright (C), 2013-2021, 广州九尾信息科技有限公司
 * @date 2021/11/12 17:57
 */
@Component
public class MyKafkaProducer {
    private static final Logger logger = LoggerFactory.getLogger(MyKafkaProducer.class);
    @Autowired
    private KafkaTemplate<String, User> kafkaTemplate;

    @Async
    public void send(String message) {
        //发送消息
        //ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(MyConstants.TOPIC, "key1", message);
        User user = new User();
        user.setName("mojo");
        user.setAge(11);
        ListenableFuture<SendResult<String, User>> future = kafkaTemplate.send(MyConstants.TOPIC, "user1", user);
        kafkaTemplate.isTransactional();
        future.addCallback(new ListenableFutureCallback<SendResult<String, User>>() {
            @Override
            public void onSuccess(SendResult<String, User> result) {
                logger.info(">>>>>producer 发送消息回调成功>>>>>>");
                RecordMetadata recordMetadata = result.getRecordMetadata();
                ProducerRecord<String, User> producerRecord = result.getProducerRecord();
                String topic = producerRecord.topic();
                Integer partition = producerRecord.partition();
                System.out.println(recordMetadata);
                System.out.println(topic);
                System.out.println(partition);
            }

            @Override
            public void onFailure(Throwable ex) {
                logger.info(">>>>>producer 发送消息回调失败>>>>>>");
            }

        });

    }

    public void myOutput(String message) {
        System.out.println("发送消息成功");
    }

    class ProducerCallBack implements Callback {

        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e != null) {
                e.printStackTrace();
            }
        }
    }


}
